package com.atguigu;

import beans.LoginEvent;
import beans.LoginFailWarning;
import org.apache.flink.api.common.eventtime.SerializableTimestampAssigner;
import org.apache.flink.api.common.eventtime.WatermarkStrategy;
import org.apache.flink.cep.CEP;
import org.apache.flink.cep.PatternSelectFunction;
import org.apache.flink.cep.PatternStream;
import org.apache.flink.cep.pattern.Pattern;
import org.apache.flink.cep.pattern.conditions.SimpleCondition;
import org.apache.flink.streaming.api.datastream.SingleOutputStreamOperator;
import org.apache.flink.streaming.api.environment.StreamExecutionEnvironment;
import org.apache.flink.streaming.api.windowing.time.Time;

import java.net.URL;
import java.time.Duration;
import java.util.List;
import java.util.Map;

/**
 * @author zkq
 * @date 2022/10/15 13:27
 */
//前面用process 不管基于时间还是基于事件都有问题 ，如果有乱序的成功 会影响报警 不够严谨 虽然也有办法解决 麻烦
public class LoginFailWithCep {
    public static void main(String[] args) throws Exception {
        StreamExecutionEnvironment env = StreamExecutionEnvironment.getExecutionEnvironment();
        env.setParallelism(1);

        //从文件中读取数据
        URL resource = LoginFailWithCep.class.getResource("/LoginLog.csv");
        String path = resource.getPath();
        SingleOutputStreamOperator<LoginEvent> LoginStream = env.readTextFile(path)
                .map(data -> {
                    String[] split = data.split(",");
                    return new LoginEvent(new Long(split[0]), split[1], split[2], Long.valueOf(split[3]));
                })
                .assignTimestampsAndWatermarks(WatermarkStrategy.<LoginEvent>forBoundedOutOfOrderness(Duration.ofSeconds(3))
                        .withTimestampAssigner(new SerializableTimestampAssigner<LoginEvent>() {
                            @Override
                            public long extractTimestamp(LoginEvent element, long recordTimestamp) {
                                return element.getTimestamp() * 1000;
                            }
                        })
                );
        /*//1.定义一个匹配模式
        //firstFail -> secondFail, within 2s
        Pattern<LoginEvent, LoginEvent> loginFailPattern = Pattern.<LoginEvent>begin("firstFail")
                .where(new SimpleCondition<LoginEvent>() {
                    @Override
                    public boolean filter(LoginEvent value) throws Exception {
                        return "fail".equals(value.getLoginState());
                    }
                })
                .next("secondFail")
                .where(new SimpleCondition<LoginEvent>() {
                    @Override
                    public boolean filter(LoginEvent value) throws Exception {
                        return "fail".equals(value.getLoginState());
                    }
                })
                .within(Time.seconds(2));

        //2.将匹配模式应用到数据流上 得到一个pattern stream
        PatternStream<LoginEvent> patternStream = CEP.pattern(LoginStream.keyBy(LoginEvent::getUserId), loginFailPattern);

        //3.检出符合匹配条件的复杂事件，进行处理
        patternStream.select(new PatternSelectFunction<LoginEvent, LoginFailWarning>() {
            @Override
            public LoginFailWarning select(Map<String, List<LoginEvent>> pattern) throws Exception {
                LoginEvent firstFailEvent = pattern.get("firstFail").iterator().next();
                LoginEvent secondFailEvent = pattern.get("secondFail").iterator().next();
                return new LoginFailWarning(firstFailEvent.getUserId(),firstFailEvent.getTimestamp(),secondFailEvent.getTimestamp(),"login fail 2 times");
            }
        })
                .print();*/
        //!!!升级版  五秒内连续3次匹配上      直接times是宽松匹配模式 加上consecutive表示严格匹配模式 within给事件序列匹配的时间 
        Pattern<LoginEvent, LoginEvent> loginFailPattern = Pattern.<LoginEvent>begin("failEvents")
                .where(new SimpleCondition<LoginEvent>() {
                    @Override
                    public boolean filter(LoginEvent value) throws Exception {
                        return "fail".equals(value.getLoginState());
                    }
                })
                .times(3)
                .consecutive()
                .within(Time.seconds(5));
        PatternStream<LoginEvent> patternStream = CEP.pattern(LoginStream.keyBy(LoginEvent::getUserId), loginFailPattern);
        patternStream.select(new PatternSelectFunction<LoginEvent, LoginFailWarning>() {
            @Override
            public LoginFailWarning select(Map<String, List<LoginEvent>> pattern) throws Exception {
                LoginEvent firstFailEvent = pattern.get("failEvents").get(0);
                LoginEvent lastFailEvent = pattern.get("failEvents").get(pattern.get("failEvents").size() - 1);
                return new LoginFailWarning(firstFailEvent.getUserId(),firstFailEvent.getTimestamp(),lastFailEvent.getTimestamp(),"login fail 3 times");
            }
        })
                .print();

        env.execute("login fail detect with cep");

    }
}
